package spark.core.scala

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * Created by ibf on 2018/2/3.
 */
object GroupSortTopN {
  def main(args: Array[String]) {
      //1、spark上下文构建
      val conf = new SparkConf()
            .setAppName("GroupSortTopN")
            .setMaster("local[*]")
      val sc = SparkContext.getOrCreate(conf)

      //2、读取数据，生成RDD
      val k = 3
      val path = "datas/groupsort.txt"
      //可以在加载数据的时候指定分区数量
      val rdd = sc.textFile(path,1)

      //3、代码逻辑，处理rdd
      //先做split，按照空格分割字段
      /**
       * 原始的处理方法：直接使用groupByKey，然后在每个key的迭代器中排序取出前K个
       */
     /* val resultRdd = rdd.map(_.split(" "))
        //字段如果长度不等于2，证明那一行数据有问题，就过滤掉
                    .filter(arr => arr.length == 2)
        //因为接下来要做groupbykey的操作，所以要做成二元组，（String,Int）
                    .map(arr => (arr(0),arr(1).toInt))
                    .groupByKey()
        //拿到每一个key对应iter
        /**
         * 如果使用map，最后结果是:List((aa,78), (aa,80), (aa,97))
                                List((bb,92), (bb,97), (bb,98))
                                List((cc,86), (cc,87), (cc,98))
         * 如果使用flatMap，最后结果是：(aa,78)(aa,80)(aa,97)(bb,92)(bb,97)....
         */

        .map(t =>{
        val key: String = t._1
        val iter: Iterable[Int] = t._2
        //对iter进行排序，取出前三个（最大值的前三个）出来
        val topKItem: List[Int] = iter.toList.sorted.takeRight(k)
        //做返回（key，iter1）（key，iter2）（key，iter3）
        topKItem.map(iter2 =>(key,iter2))
      })
  */
      /**
       * 优化一：
       * 当我们使用groupByKey的时候，有可能会出现问题
       * 当某一个key数据量特别多的时候，会导致数据倾斜（某个分区/task处理结果非常慢）
       * 影响最后的结果输出
       *    使用两阶段聚合==》解决数据倾斜问题
       */
      /**
       * 第一次聚合：将相同的((随机数，key)，value)的value聚合，排序取出前三个
       */
      /*val resultRdd = rdd.map(_.split(" "))
                .filter(arr => arr.length ==2)
        //要将Key加上一个前缀，相当于将key做一次散列
        //原来：(aa,98)  ==> ((23,aa),98)
                .map(arr =>((Random.nextInt(100),arr(0)),arr(1).toInt))
                .groupByKey()
                .flatMap{
        case((_,key),iter) => {
          //对iter进行排序然后获取数值最大的前K个数据
          val topKList = iter.toList.sorted.takeRight(k)
          topKList.map(iter => (key,iter))
        }
      }
      //再做一次全局的聚合groupByKey
        .groupByKey()
        .flatMap{
        case(key,iter) => {
          val topKList2 = iter.toList.sorted.takeRight(k)
          topKList2.map(iter => (key,iter))
        }
      }*/

      /**
       * 优化二：使用默认自带combiner的算子，来解决数据倾斜的问题
       *  例如：reduceByKey   aggregateByKey
       */
      /**
       *  def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int)(seqOp: (U, V) => U,
            combOp: (U, U) => U): RDD[(K, U)] = self.withScope {
           aggregateByKey(zeroValue, new HashPartitioner(numPartitions))(seqOp, combOp)
            }
        用法：
          zeroValue：传入计算时的初始值
          seqOp：combiner时候的运算（局部聚合操作）
          combOp：reduce时候的运算（全局聚合的操作）
       */
      val resultRdd: RDD[(String, ArrayBuffer[Int])] = rdd.map(_.split(" "))
        .filter(arr => arr.length ==2)
        .map(arr => (arr(0),arr(1).toInt))
        //aggregateByKey可以使用函数的柯里化给定函数的初始值
        //ArrayBuffer[Int]()给定一个空的可变长的数据作为初始值
        .aggregateByKey(ArrayBuffer[Int]())(
          (u,v) => {
            val key: ArrayBuffer[Int] = u
            val i: Int = v
            u +=v
            u.sorted.takeRight(k)
          },
          /**
           *  经过了seqOp的运算，相同key可能会生成多个ArrayBuffer[Int]()
           *  所以要再定义一个方法，来将相同key的多个ArrayBuffer[Int]()做一次运算
           *  取出前三个
           */
          (u1,u2) =>{
            val arr1: ArrayBuffer[Int] = u1
            val arr2: ArrayBuffer[Int] = u2
            //加起来，最多最多也就6个值
            u1 ++= u2
            u1.sorted.takeRight(k)
          }
        )


      //4、处理完数据，做结果打印
      resultRdd.foreachPartition(iter => {
        //iter是迭代器，所以只能使用一次，所以不能使用iter直接求长度
        //val list: List[List[(String, Int)]] = iter.toList
        //println("每个分区的长度：==============="+list.size)
        var i = 0
        iter.foreach(t => {
          i+=1
          println(t)
        })
        println(s"迭代第${i}次")
      })

  }
}
